package com.mgzf.bs.uploadCenter.service.impl;


import com.mgzf.bs.uploadCenter.service.IkafkaService;
import com.mgzf.bs.uploadCenter.service.task.IMsgTask;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * Created by mike on 18/1/18.
 */
@Service
public class KafkaServiceImpl implements IkafkaService {
    private final Logger LOGGER = Logger.getLogger(KafkaServiceImpl.class);

    private Boolean recvStopSign = true;

    @Override
    public Boolean getRecvStopSign() {
        return recvStopSign;
    }

    @Override
    public void setRecvStopSign(Boolean jump) {
        this.recvStopSign = jump;

    }

    /**房态同步消息发送频率*/
    @Value("${sysconfig.kafka.sync_roomstatus_interval_in_sec:2}")
    private Integer syncRoomStatusIntervalInSec;

    @Value("${sysconfig.kafka.zk-connect}")
    private String zkServer;

    @Value("${sysconfig.kafka.group:myTestGroup}")
    private String groupId;

    @Override
    public void receive(String topic, final IMsgTask msgTask) {
        Properties props = new Properties();
        props.put("zookeeper.connect", zkServer);
        props.put("group.id", groupId);
        props.put("zookeeper.session.timeout.ms", "14000");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "1000");
        props.put("auto.commit.enable", "true");
        ConsumerConfig conf = new ConsumerConfig(props);
        ConsumerConnector cc = Consumer.createJavaConsumerConnector(conf);
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();

        topicCountMap.put(topic, 4);
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = cc.createMessageStreams(topicCountMap);
        List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);
        for (final KafkaStream<byte[], byte[]> stream : streams) {
            new Thread() {
                public void run() {
                    ConsumerIterator<byte[], byte[]> it = stream.iterator();
                    while (getRecvStopSign() && it.hasNext()) {
                        String msg = new String(it.next().message());
                        try {
                            msgTask.execute(msg);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    LOGGER.info("====接收到线程终止信号，执行结束，信号状态："+ getRecvStopSign());
                }
            }.start();

        }
        LOGGER.info("kafka Consumer 准备接收消息：");

    }
}
